package com.etl.component.common;

import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.LocalDateTimeUtil;
import cn.hutool.core.util.StrUtil;
import com.component.api.AbstractComponent;
import com.component.api.excepiton.ComponentException;
import com.component.api.model.ProcessResult;
import com.component.api.model.data.SetData;
import com.component.api.model.data.StaticRowData;
import com.component.api.model.param.TableColumn;
import com.common.utils.constant.Constants;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.LocalTimeTypeInfo;
import org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.StringUtils;

import java.math.BigDecimal;
import java.math.BigInteger;
import java.sql.Time;
import java.sql.Timestamp;
import java.sql.Types;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * 描述：
 * Flink组件抽象类
 * @author xianggj
 * @Date 2021/11/25 11:20
 **/
public abstract class AbstractFlinkComponent extends AbstractComponent {

    /**
     *
     * 保证上下文统一
     * 一个流程算子为一个环境 避免相互影响
     */
    protected static final Map<String, ExecutionEnvironment> envMap = new ConcurrentHashMap<>();

    public ExecutionEnvironment getEnv() {
        return envMap.get(getFlowKey());
    }

    /**
     * 用完清楚掉
     */
    public static void clearEnv(String flowKey) {
        envMap.remove(flowKey);
    }

    /**
     * 用完清楚掉
     * @param flowKey
     */
    public static void putEnv(String flowKey, ExecutionEnvironment env) {
        envMap.put(flowKey, env);
    }


    /**
     * 字符串类型转BasicTypeInfo
     * @param type
     */
    public TypeInformation typeToBasicTypeInfo(String type) {
        type = StrUtil.isBlank(type) ? "string" : type.toLowerCase();
        switch (type) {
            case "int":
            case "integer":
            case "int2":
            case "int4":
            case "smallint":
            case "tinyint":
                return BasicTypeInfo.INT_TYPE_INFO;
            case "int8":
            case "long":
            case "bigint":
                return BasicTypeInfo.LONG_TYPE_INFO;
//不支持bigint类型
//                return BasicTypeInfo.BIG_INT_TYPE_INFO;
            case "float":
            case "float2":
            case "float4":
                return BasicTypeInfo.FLOAT_TYPE_INFO;
            case "float8":
            case "double":
                return BasicTypeInfo.DOUBLE_TYPE_INFO;
            case "time":
                return SqlTimeTypeInfo.TIME;
            case "date":
                return SqlTimeTypeInfo.DATE;
            case "datetime":
                return LocalTimeTypeInfo.LOCAL_DATE_TIME;
            case "timestamp":
            case "timestamp(6)":
                return SqlTimeTypeInfo.TIMESTAMP;
            case "boolean":
            case "bool":
                return BasicTypeInfo.BOOLEAN_TYPE_INFO;
            case "byte":
                return BasicTypeInfo.BYTE_TYPE_INFO;
            case "short":
                return BasicTypeInfo.SHORT_TYPE_INFO;
//            case "character":
//            case "char":
//                return BasicTypeInfo.CHAR_TYPE_INFO;
            case "decimal":
            case "bigdecimal":
            case "numeric":
                return BasicTypeInfo.BIG_DEC_TYPE_INFO;
            default:
                return BasicTypeInfo.STRING_TYPE_INFO;
        }

    }

    /**
     * 字符串类型转Int
     * 这个类里面有现成方法 但是引用不了
     * JDBCTypeUtil
     *        HashMap<TypeInformation<?>, Integer> m = new HashMap();
     *         m.put(BasicTypeInfo.STRING_TYPE_INFO, 12);
     *         m.put(BasicTypeInfo.BOOLEAN_TYPE_INFO, 16);
     *         m.put(BasicTypeInfo.BYTE_TYPE_INFO, -6);
     *         m.put(BasicTypeInfo.SHORT_TYPE_INFO, 5);
     *         m.put(BasicTypeInfo.INT_TYPE_INFO, 4);
     *         m.put(BasicTypeInfo.LONG_TYPE_INFO, -5);
     *         m.put(BasicTypeInfo.FLOAT_TYPE_INFO, 7);
     *         m.put(BasicTypeInfo.DOUBLE_TYPE_INFO, 8);
     *         m.put(SqlTimeTypeInfo.DATE, 91);
     *         m.put(SqlTimeTypeInfo.TIME, 92);
     *         m.put(SqlTimeTypeInfo.TIMESTAMP, 93);
     *         m.put(BasicTypeInfo.BIG_DEC_TYPE_INFO, 3);
     *         m.put(PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO, -2);
     */
    public int typeToInt(TypeInformation typeInfo) {

        typeInfo = typeInfo == null?BasicTypeInfo.STRING_TYPE_INFO: typeInfo;
        if (BasicTypeInfo.STRING_TYPE_INFO.equals(typeInfo)){
            return Types.VARCHAR;
        } else if (BasicTypeInfo.INT_TYPE_INFO.equals(typeInfo)){
            return Types.INTEGER;
        } else if (BasicTypeInfo.DOUBLE_TYPE_INFO.equals(typeInfo)){
            return Types.DOUBLE;
        } else if (BasicTypeInfo.CHAR_TYPE_INFO.equals(typeInfo)){
            return Types.CHAR;
        } else if (SqlTimeTypeInfo.TIME.equals(typeInfo)){
            return Types.TIME;
        } else if (SqlTimeTypeInfo.DATE.equals(typeInfo)){
            return Types.DATE;
        } else if (LocalTimeTypeInfo.LOCAL_DATE_TIME.equals(typeInfo)){
            return Types.TIMESTAMP_WITH_TIMEZONE;
        } else if (SqlTimeTypeInfo.TIMESTAMP.equals(typeInfo)){
            return Types.TIMESTAMP;
        } else if (BasicTypeInfo.LONG_TYPE_INFO.equals(typeInfo)||BasicTypeInfo.BIG_INT_TYPE_INFO.equals(typeInfo)){
            return Types.BIGINT;
        } else if (BasicTypeInfo.FLOAT_TYPE_INFO.equals(typeInfo)||BasicTypeInfo.SHORT_TYPE_INFO.equals(typeInfo)){
            return Types.REAL;
        } else if (BasicTypeInfo.BOOLEAN_TYPE_INFO.equals(typeInfo)){
            return Types.BOOLEAN;
        } else if (BasicTypeInfo.BYTE_TYPE_INFO.equals(typeInfo)){
            return Types.BINARY;
        } else if (BasicTypeInfo.BIG_DEC_TYPE_INFO.equals(typeInfo)){
            return Types.DECIMAL;
        } else {
            return Types.VARCHAR;
        }
    }


    /**
     * 值转换
     * @param oldVal 原值
     * @param typeInfo 需要转换的类型
     * @return 异常自行处理
     */
    public static Object covert(Object oldVal, TypeInformation typeInfo){
        if (oldVal == null) {
            return null;
        }
        String valStr = oldVal.toString();
        if (BasicTypeInfo.STRING_TYPE_INFO.equals(typeInfo)){
            return valStr;
        } else if (BasicTypeInfo.INT_TYPE_INFO.equals(typeInfo)){
            //double 四舍五入
            return (int) Math.round(Double.valueOf(valStr));
        } else if (BasicTypeInfo.DOUBLE_TYPE_INFO.equals(typeInfo)){
            return Double.valueOf(valStr);
        } else if (BasicTypeInfo.CHAR_TYPE_INFO.equals(typeInfo)){
            return valStr.charAt(0);
        } else if (SqlTimeTypeInfo.TIME.equals(typeInfo)){
            DateTime parse = DateUtil.parse(valStr);
            return new Time(parse.getTime());
        } else if (SqlTimeTypeInfo.DATE.equals(typeInfo)){
            DateTime parse =  DateUtil.parse(valStr);
            return  new java.sql.Date(parse.getTime());
        } else if (LocalTimeTypeInfo.LOCAL_DATE_TIME.equals(typeInfo)){
            DateTime parse =  DateUtil.parse(valStr);
            return LocalDateTimeUtil.of(parse);
        } else if (SqlTimeTypeInfo.TIMESTAMP.equals(typeInfo)){
            DateTime parse = DateUtil.parse(valStr);
            return new Timestamp(parse.getTime());
        } else if (BasicTypeInfo.LONG_TYPE_INFO.equals(typeInfo)){
            return Math.round(Double.valueOf(valStr));
        } else if (BasicTypeInfo.BIG_INT_TYPE_INFO.equals(typeInfo)){
            return new BigInteger(StrUtil.subBefore(valStr, ".", true));
        } else if (BasicTypeInfo.FLOAT_TYPE_INFO.equals(typeInfo)){
            return Float.valueOf(valStr);
        } else if (BasicTypeInfo.SHORT_TYPE_INFO.equals(typeInfo)){
            return (short)Math.round(Double.valueOf(valStr));
        } else if (BasicTypeInfo.BOOLEAN_TYPE_INFO.equals(typeInfo)){
            return Boolean.valueOf(valStr);
        } else if (BasicTypeInfo.BYTE_TYPE_INFO.equals(typeInfo)){
            return Byte.valueOf(valStr);
        } else if (BasicTypeInfo.BIG_DEC_TYPE_INFO.equals(typeInfo)){
            return new BigDecimal(valStr);
        }
        return oldVal;
    }

    /**
     * 执行动作
     *  如果一个dataset多次执行会出现错误
     * @param result
     */
    protected synchronized ProcessResult execute(DataSet<Row> result, List<TableColumn> columns, String msg) {
        ProcessResult processResult = new ProcessResult();
        if (getNeedExec()){
            try {
                List<Row> collect = result.collect();
                List<String> rows = collect.stream().map(e ->
                        toQuotedListString(e)).collect(Collectors.toList());
                processResult.addAbstractTableData(getRule(), new StaticRowData(rows, columns));
                System.out.println("总共执行了"+collect.size() + "条数据");
            } catch (Exception e) {
                e.printStackTrace();
                throw new ComponentException(msg+"时出现错误：", e);
            }
        }else {
            processResult.addAbstractTableData(getRule(), new SetData(result, columns));
        }
        return processResult;
    }

    public static String toQuotedListString(Object[] values) {
        return (String) Arrays.stream(values).filter(Objects::nonNull).map((v) -> {
            return Constants.DOUBLE_QUOTATION +v.toString()+ Constants.DOUBLE_QUOTATION;
        }).collect(Collectors.joining(Constants.COMMA));
    }

    /**
     * row字符串预览
     * @param row
     * @return
     */
    public static String toQuotedListString(Row row) {

        Object[] values = new Object[row.getArity()];
        for(int i = 0; i < row.getArity(); i++) {
            String string = StringUtils.arrayAwareToString(row.getField(i));
            values[i] = StrUtil.removeAll(string, Constants.DUBLE_COMMA);
        }
        return toQuotedListString(values);
    }

}
